package org.example.aggregation;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.example.flink.pojo.WaterSensor;

import java.util.Arrays;

/**
 * @Author: tang
 * @Description: 增量聚合函数 ReduceFunction
 * @Date 2025/2/16 16:55
 */
public class FlinkAggregation {

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.fromCollection(
                Arrays.asList(
                        new WaterSensor("1", "11", 10),
                        new WaterSensor("1", "10", 11),
                        new WaterSensor("2", "22", 20),
                        new WaterSensor("3", "33", 30)
                )
        )
                .keyBy((KeySelector<WaterSensor, String>) value -> value.id)
//                .minBy("vc")
                .reduce((value1, value2) -> {
                    return new WaterSensor(value1.id, value1.vt, (value1.vc + value2.vc));
                })
                .print()
        ;

        environment.execute();

    }

}
